package Spark_Scala_RDD

import org.apache.spark.{SparkConf, SparkContext}

object HelloSpark {

  def main(args: Array[String]): Unit = {

    System.setProperty("hadoop.home.dir","D:\\hadoop-2.7.3\\hadoop-2.7.3")
    System.setProperty("HADOOP_USER_NAME","root")
    /*spark:计算框架 *
    RDD:数据类型 全称：弹性分布式数据集
                    弹性： -计算逻辑:很长+很短    --容错性+可恢复性
                    分布式： -RDD可分布在各个节点/机器/电脑  -分布式存储
                    数据集： 数据类型
        五大特性：
               1.一个分区： 计算时 在分区里partition计算
               2.一个依赖列表： rdd.map().reduceByKey() --map()映射必须和前面连接
               3.一个函数： 函数式编程（匿名函数）/也可以有名字
               4.可选的分区器
               5.位置感知调度： 对于k-v类型的 散列分区器，同时RDD所在的每个分区都被记录了其所在的节点信息

     */

    //RDD的创建
    //添加spark-core依赖
    //通过 Driver (SparkContext --执行函数？-开启运用方法的入口-相当于Java/Scala中的main函数)
    //通过 读取文件 （hdfs,文件txt,hive数据仓库）

    val conf = new SparkConf().setMaster("local").setAppName("Hellospark")
    val  sc = new SparkContext(conf)
    //array
    val arr = Array(1,2,3,4,5)
    val arrRDD = sc.parallelize(arr)
    //求和
    //println(arrRDD.sum())

    //spark语言强大-但凡写两行代码不是好代码

    //词频统计
    val word = Array("hello spark","hello scala","hello java")
    //mr map reduce main  ??????

    //flatMap 算子 ，rdd里边有一个元素-扁平化处理   ((1,1.1))  ==>   (1,1.1)
    //("hello spark","hello scala","hello java")
    //flatMap (hello   spark   hello   scala)
    //map(hello,1)  (spark,1)   (hello,1)   (scala,1)...
    //reduceByKey 按照Key值是否相同聚合  (hello,3) (spark,1)...
    //foreach循环遍历输出

    //spark函数式编程
    //spark-“_”--是通配符---可以接收RDD集合中的每个元素

    //parallelize--使返回值数据类型变为RDD         flatMap里传的参数是一个函数 --对RDD里面的每一个元素进行划分压平操作
    //          --把word数组转换为RDD类型          对数据源进行切分-每一个元素切分之后以括号的形式返回 "hello spark" -split- ==>RDD((hello , spark),(hello , scala)) -flatMap- ==> RDD(hello , spark , hello , scala)
    /*sc.parallelize(word).flatMap((x:String) => {x.split("")})
      .map((x:String) => {(x,1)}).reduceByKey((x:Int,y:Int) => {x+y}).foreach(println(_))*/
     //map--使RDD变成( (hello,1)           将通过map变成后的形式的value值进行相加
    // (spark,1) (hello,1)...)的形式       按照key是否相同,（如果相同+1）在reduce端进行聚合 -==>RDD(("hello",3),("spark",1)) --- 即完成词频统计

    //上面是复杂的写法
    //变形--
    // 1.括号去掉(最外围-中括号)-2.类型去掉((x:String)-(x))-3.参数小括号去掉((x)-x)-4.参数去掉换成通配符_
    //sc.parallelize(word).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println(_))

    //多行
    /*val a = sc.parallelize(word).flatMap(_.split(" "))
    val b = a.map((_,1))
    val c = b.reduceByKey(_+_)
    c.foreach(println(_))*/

    //println(a) //如果像这样写一半运行-没有结果---并不是因为编码没有写完无法运行
    //而是因为：
    //spark算子--一种函数逻辑  flatMap: 一个transformation算子，所以无法运行出结果
    //                      map: 变成(k,v) => key值+value值=1的形式(hello,1) --也可以单划出来成一类算子
    //                      RDD.sum(): 加法 || count(): 统计数量
    //spark算子分为两大类：transforma算子：无法运行出结果 --- flatMap()   map()   reduceByKey()
    //                               进行计算的时候，保存了计算的逻辑，没有进行结果的运算
    //                  action算子：可以出结果 --- foreach()  sum()   count()
    //                             编码只有遇到action类型算子，才进行结果运算
    //如果a.foreach(printIn())---运行出来就会有结果
    /*println("1111111111111")
    a.foreach(println(_))*/

    //读取文件中的内容进行词频统计
    // val rd = sc.textFile(文件路径) --> rd.flatMap-map-reduceByKey-foreach

    //filter过滤算子用法
    /*sc.parallelize(word).flatMap(_.split(" "))
      .filter(item => item != "java")
      .map((_,1))
      .reduceByKey(_+_)
      .foreach(println(_))*/


    //Git



  }

}
